package com.hs.learn.config;

import com.hs.learn.changestream.DocumentMessageListener;
import com.mongodb.ReadConcern;
import com.mongodb.ReadPreference;
import com.mongodb.TransactionOptions;
import com.mongodb.WriteConcern;
import com.mongodb.client.model.changestream.FullDocument;
import org.bson.Document;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.MongoTransactionManager;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.mongodb.core.query.Criteria;

import java.util.concurrent.*;

/**
 * @Description: mongodb的配置类
 * @Author 胡尚
 * @Date: 2024/7/31 14:52
 */
@Configuration
public class MongodbConfig {

    /**
     * 指定readPreference、readConcern、writeConcern
     */
    @Bean
    public MongoTransactionManager transactionManager(MongoDatabaseFactory factory) {
        TransactionOptions transactionOptions = TransactionOptions.builder()
                .readPreference(ReadPreference.primary())
                .readConcern(ReadConcern.MAJORITY)
                .writeConcern(WriteConcern.MAJORITY)
                .build();

        return new MongoTransactionManager(factory, transactionOptions);
    }

    /**
     * change stream的配置
     *
     * @param template                引入mongodb的依赖后，内置的bean对象
     * @param documentMessageListener 自定义的messageListener
     * @return
     */
    @Bean
    public MessageListenerContainer messageListenerContainer(MongoTemplate template, DocumentMessageListener documentMessageListener) {
        // 创建一个线程池
        Executor executor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(5));

        // 创建一个MessageListenerContainer对象
        MessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(template, executor) {
            @Override
            public boolean isAutoStartup() {
                return true;
            }
        };

        ChangeStreamRequest<Document> changeStreamRequest = ChangeStreamRequest.builder(documentMessageListener)
                // 需要监听的集合名
                .collection("emp")
                //过滤需要监听的操作类型，可以根据需求指定过滤条件
                .filter(Aggregation.newAggregation(Aggregation.match(
                        Criteria.where("operationType").in("insert", "update", "delete"))))
                //不设置时，文档更新时，只会发送变更字段的信息，设置UPDATE_LOOKUP会返回文档的全部信息
                .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
                .build();

        messageListenerContainer.register(changeStreamRequest, Document.class);

        return messageListenerContainer;
    }
}
